package com.lac.component.rabbitmq;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.lac.component.model.MessageInfo;
import com.lac.component.rabbitmq.dead.DelayedConfig;
import com.lac.component.rabbitmq.dead.DeadLetterConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@Component
public class MsgProducer  {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    //由于rabbitTemplate的scope属性设置为ConfigurableBeanFactory.SCOPE_PROTOTYPE，所以不能自动注入
    @Autowired
    private RabbitTemplate rabbitTemplate;
    private RedisTemplate redisTemplate;

    public MsgProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }


    public void sendMsg(String goodsId,String content){
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        //Fanout 就是我们熟悉的广播模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。
        //rabbitTemplate.convertAndSend(RabbitConfig.FANOUT_EXCHANGE,content);
        //把消息放入ROUTINGKEY_A对应的队列当中去，对应的是队列A
        //rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A,RabbitConfig.ROUTINGKEY_A,content,correlationId);
        //传输对象
        Map mp = new HashMap(1024);
        mp.put("goodsId",goodsId);
        mp.put("reduce",Integer.valueOf(content));
        mp.put("uid",UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A,RabbitConfig.ROUTINGKEY_A,mp,correlationId);
        //rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A,RabbitConfig.ROUTINGKEY_A,user,correlationId);

    }


    // 死信队列
    public void sendDelayedMessage(String message, int delaySeconds) {
//        System.out.println("[" + LocalDateTime.now() + "] 发送延迟消息: " + message + ", 延迟: " + delaySeconds + "秒");
        rabbitTemplate.convertAndSend(DeadLetterConfig.WORK_EXCHANGE, DeadLetterConfig.WORK_ROUTING_KEY, message, msg -> {
            msg.getMessageProperties().setExpiration(String.valueOf(delaySeconds * 1000));
            return msg;
        });
    }

    // 延时队列
    public void sendDelayedMessage(String message) {
//        System.out.println("[" + LocalDateTime.now() + "] 延迟10s发送消息: " + message);
        rabbitTemplate.convertAndSend(DelayedConfig.DELAYED_EXCHANGE_NAME, DelayedConfig.DELAYED_ROUTING_KEY, message, messagePostProcessor -> {
            messagePostProcessor.getMessageProperties().setDelay(100000); // 设置延迟时间，单位毫秒
            messagePostProcessor.getMessageProperties().setContentType("application/json");
            return messagePostProcessor;
        });
    }
    /**
     * 根据id，将消息顺序发送到对应的队列
     * @param id 业务id
     * @param msg 业务信息
     */
    public void send(int id, String msg) throws JsonProcessingException {
        MessageInfo message = new MessageInfo();
        message.setId(id);
        message.setMessage(msg);
        rabbitTemplate.convertAndSend("direct_seq", String.valueOf(id % 2 + 1), message, messagePostProcessor -> {
            messagePostProcessor.getMessageProperties().setDelay(100000); // 设置延迟时间，单位毫秒
            messagePostProcessor.getMessageProperties().setContentType("application/json");
            return messagePostProcessor;
        });
    }

    public void convertAndSend(String  exchange,String routingKey,MessageInfo message) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message, messagePostProcessor -> {
            messagePostProcessor.getMessageProperties().setDelay(100000); // 设置延迟时间，单位毫秒
            messagePostProcessor.getMessageProperties().setContentType("application/json");
            return messagePostProcessor;
        });
    }

    private int partitionCount = 3;

    // 初始化交换和队列（确保运行时创建）
    @PostConstruct
    public void setup() {
        // 交换
        rabbitTemplate.execute(channel -> {
            channel.exchangeDeclare("partition_exchange", "topic", true);
            // 队列绑定在配置类中已实现
            return null;
        });
    }

    /**
     * 计算消息路由键（根据某个属性哈希）
     */
    public void sendMessage(MessageInfo message, String key) {
        int partitionIndex = Math.abs(key.hashCode()) % partitionCount;
        String routingKey = "partition." + partitionIndex;
        rabbitTemplate.convertAndSend("partition_exchange", routingKey, message);
//        System.out.println("Sent message to partition: " + partitionIndex);
    }



}
